package com.qqt.csr.im.mq;

import com.alibaba.fastjson2.JSONObject;
import com.google.common.collect.Lists;
import com.qqt.csr.common.utils.CacheKeyUtil;
import com.qqt.csr.common.utils.JsonUtil;
import com.qqt.csr.common.utils.RedissonLockUtil;
import com.qqt.csr.im.service.sensitiveword.SensitiveWordRepoService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author:
 * @Date:
 * @Description: 导入敏感词
 **/
@Slf4j
@Service
@RocketMQMessageListener(
        consumerGroup = "${rocketmq.consumer.group}_import_sw",
        topic = "import_im_sw_topic",
        tlsEnable = "${rocketmq.consumer.tlsEnable}"
)
public class ImportSensitiveWordListener implements RocketMQListener<String> {
    @Autowired
    private SensitiveWordRepoService sensitiveWordRepoService;
    @Autowired
    private RedissonLockUtil redissonLockUtil;

    @Override
    public void onMessage(String data) {
        try {
            JSONObject jsonObject = JsonUtil.toObject(data, JSONObject.class);
            if (jsonObject == null) {
                return;
            }
            String seqId = jsonObject.getString("seqId");
            String tenantId = jsonObject.getString("tenantId");
            Long creatorId = jsonObject.getLong("creatorId");
            if (StringUtils.isBlank(tenantId) || creatorId == null) {
                return;
            }
            Map<String, Set<String>> dataMap = Optional.ofNullable(jsonObject.getList("rowDataList", JSONObject.class)).orElseGet(Lists::newArrayList)
                    .stream()
                    .filter(Objects::nonNull)
                    .filter(item -> StringUtils.isNotBlank(item.getString("0")) && StringUtils.isNotBlank(item.getString("1")))
                    .collect(Collectors.groupingBy(item -> item.getString("0").trim(),
                            Collectors.mapping(item -> item.getString("1").trim(), Collectors.toSet())));
            String cacheKey = String.format(CacheKeyUtil.IM.IMPORT_SW_MQ_LOCK, seqId);
            redissonLockUtil.distributedLock(() -> {
                sensitiveWordRepoService.batchSave(tenantId, creatorId, dataMap);
            }, cacheKey, 0, 30, TimeUnit.MINUTES);
        } catch (Exception e) {
            log.error("导入敏感词失败！data:{}", data, e);
        }
    }
}
